package com.shujia;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.util.ArrayList;
import java.util.List;


/**
 * 需求： 传入两个参数，第一个参数为字符串 第二参数为分隔符 可以实现将字符串进行分割并且展示成多行结果
 *
 *  SELECT myexplode('hello,hello,hello',',')
 *  返回：
 *  hello
 *  hello
 *  hello
 *
 * add jar /usr/local/soft/jars/hiveCode-1.0.jar;
 *
 *  create temporary function myexplode as 'com.shujia.MyUDTF';
 *
 */
public class MyUDTF extends GenericUDTF {

    /**
     * initialize 可以手动添加，在该函数中，可以实现对参数的类型长度的判断，
     *      同时UDTF函数中需要在该函数中，设置新增列的类型及名称
     */

    /**
     *
     * @param argOIs
     * @return
     * @throws UDFArgumentException
     */
    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs)
            throws UDFArgumentException {

        List<? extends StructField> allStructFieldRefs = argOIs.getAllStructFieldRefs();
        // 参数长度为 2
        if(allStructFieldRefs.size() != 2){
            throw new UDFArgumentLengthException("当前参数的长度要求限制为两个，第一个参数为字符串，第二个参数为分隔符");
        }

        // 对参数类型进行判断
        for (int i = 0; i < allStructFieldRefs.size(); i++) {
            // 抛出类型异常
            if (!allStructFieldRefs.get(i)
                    .getFieldObjectInspector()
                    .getCategory()
                    .equals(ObjectInspector.Category.PRIMITIVE)) {
               throw new UDFArgumentTypeException(i+1,"当前第"+(i+1)+"个参数类型不匹配");
            }
        }

        // 表示返回值结果的列名称
        ArrayList<String> fieldNames = new ArrayList<String>();
        fieldNames.add("words");

        // 对应列的类型
        ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
//        fieldOIs.add(PrimitiveObjectInspectorFactory.w);

        // 将返回值的列描述信息进行打包并创建StructObjectInspector返回
        return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
    }


    /**
     * 用于编写处理逻辑
     * @param args 传入参数的数组 'hello,hello,hello',','
     * @throws HiveException
     */
    @Override
    public void process(Object[] args) throws HiveException {
        // 对传入的数据进行切分
        String data = args[0].toString();
        String seg = args[1].toString();
        String[] splitRes = data.split(seg);

        // 将切分的结果进行循环写出  col变量中存储的数据为一行数据
        // 如果在initialize中对应设置的输出列数据类型为WritableString 那么当前行数据类型需要变更为 Text
        String[] col = new String[1];
        for (String split : splitRes) {
            col[0] =  split;
            // 将切分后的结果进行写出，每执行一次forward表示输出一行数据
            forward(col);
        }
    }

    @Override
    public void close() throws HiveException {
        // TODO: NoThing todo...
    }
}
